package routes

import (
	"encoding/json"
	"log"
	"net/http"
	"task_queue/global"
	"task_queue/internal/handlers" // 导入 handlers 包
	"task_queue/internal/middleware"
	"task_queue/internal/models"
	"task_queue/internal/plugin/mq"
	"task_queue/internal/response" // 导入 response 包，用于根路径
	"task_queue/internal/services"
	"task_queue/internal/utils"

	"github.com/spf13/viper"

	"github.com/gin-gonic/gin"
	"go.uber.org/zap"
	"gorm.io/datatypes"
	"gorm.io/gorm"
)

// RegisterRoutes 注册应用的所有路由
func RegisterRoutes(r *gin.Engine, db *gorm.DB) {
	// 健康检查或根路径
	r.GET("/", func(c *gin.Context) {
		response.Success(c, gin.H{"message": "Task Queue API is running"})
	})
	// 添加获取队列注册状态的 GET 接口
	r.GET("/queue-status", func(c *gin.Context) {

		queueStatus := updateQueueStatus(global.GlobalP.DB, global.GlobalP.Logger)
		response.Success(c, queueStatus)
	})
	r.StaticFS("/downloads", http.Dir("pkg/static/mq_docker_compose"))
	r.POST("/login", Login)

	// 初始化消息队列插件
	mqPlugins := initMQPlugins(db)

	authKeyHandler := handlers.NewAuthKeyHandler(db)

	mqConfigHandler := handlers.NewMQConfigHandler(db, mqPlugins)
	messagePushHandler := handlers.NewMessageHandler(db, mqPlugins)
	// API V1 分组
	apiV1 := r.Group("/api/v1")
	{
		// AuthKeys 路由组
		authKeys := apiV1.Group("/auth-keys").Use(middleware.TokenMiddleWare)
		{
			authKeys.POST("", authKeyHandler.CreateAuthKey)       // POST /api/v1/auth-keys
			authKeys.GET("/:id", authKeyHandler.GetAuthKey)       // GET /api/v1/auth-keys/:id
			authKeys.GET("", authKeyHandler.ListAuthKeys)         // GET /api/v1/auth-keys
			authKeys.PUT("/:id", authKeyHandler.UpdateAuthKey)    // PUT /api/v1/auth-keys/:id
			authKeys.DELETE("/:id", authKeyHandler.DeleteAuthKey) // DELETE /api/v1/auth-keys/:id
		}
		// MQConfig 路由组
		mqConfigs := apiV1.Group("/mq-configs").Use(middleware.TokenMiddleWare)
		{
			mqConfigs.POST("", mqConfigHandler.CreateMQConfig)       // POST /api/v1/mq-configs
			mqConfigs.GET("/:id", mqConfigHandler.GetMQConfig)       // GET /api/v1/mq-configs/:id
			mqConfigs.GET("", mqConfigHandler.ListMQConfigs)         // GET /api/v1/mq-configs
			mqConfigs.PUT("/:id", mqConfigHandler.UpdateMQConfig)    // PUT /api/v1/mq-configs/:id
			mqConfigs.DELETE("/:id", mqConfigHandler.DeleteMQConfig) // DELETE /api/v1/mq-configs/:id
		}

		// 生产者
		products := apiV1.Group("/product")
		// Use(middleware.AuthMiddleware())
		{
			products.POST("/json", messagePushHandler.PushMessage)
			products.GET("/messages", messagePushHandler.GetPendingMessages)
			products.GET("/messages_details", messagePushHandler.GetMessagesByStatus)
			products.POST("/consume", messagePushHandler.ConsumeMessages)
			products.GET("/topics", messagePushHandler.ListTopics)
			products.DELETE("/topic", messagePushHandler.DeleteTopic)
		}

	}

}

//=================================================================================================================

func updateQueueStatus(db *gorm.DB, logger *zap.Logger) QueueRegistrationStatus {

	scheduler := setupScheduler(db, logger)
	return QueueRegistrationStatus{
		RabbitMQRegistered:     scheduler.RabbitMQPlugin != nil,
		KafkaRegistered:        scheduler.KafkaPlugin != nil,
		NSQRegistered:          scheduler.NSQPlugin != nil,
		RocketMQRegistered:     scheduler.RocketMQPlugin != nil,
		RedisRegistered:        scheduler.RedisPlugin != nil,
		ChannelQueueRegistered: scheduler.ChannelQueuePlugin != nil,
	}
}

// 定义队列注册状态结构体
type QueueRegistrationStatus struct {
	RabbitMQRegistered     bool `json:"rabbitmq_registered"`
	KafkaRegistered        bool `json:"kafka_registered"`
	NSQRegistered          bool `json:"nsq_registered"`
	RocketMQRegistered     bool `json:"rocketmq_registered"`
	RedisRegistered        bool `json:"redis_registered"`
	ChannelQueueRegistered bool `json:"channel_queue_registered"`
}

func setupScheduler(db *gorm.DB, logger *zap.Logger) services.Scheduler {
	scheduler := services.Scheduler{} // 初始化空的调度器

	// 初始化内存队列插件
	channelQueueConfig := map[string]interface{}{
		"buffer_size":   1000,
		"default_topic": "default",
		"db":            db, // 传递数据库连接
	}
	plugin, err := mq.NewMQPlugin(mq.ChannelQueue, channelQueueConfig)
	if err == nil {
		// 使用类型断言将接口转换为具体类型
		scheduler.ChannelQueuePlugin = plugin
		logger.Info("Channel Queue plugin initialized.")
	} else {
		logger.Error("Failed to initialize Channel Queue plugin", zap.Error(err))
	}

	// 从数据库获取其他消息队列配置并初始化
	var mqConfigs []models.MQConfig
	if err := db.Find(&mqConfigs).Error; err == nil {
		for _, mqConfig := range mqConfigs {
			config := map[string]interface{}{}

			switch models.MQType(mqConfig.PluginType) {
			case models.RabbitMQ:
				config["conn_string"] = mqConfig.BrokerURLs
				plugin, err := mq.NewMQPlugin(mq.RabbitMQ, config)
				if err == nil {
					// 使用类型断言
					scheduler.RabbitMQPlugin = plugin
					logger.Info("RabbitMQ plugin initialized.")
				} else {
					logger.Error("Failed to initialize RabbitMQ plugin", zap.Error(err))
				}

			case models.Kafka:
				config["brokers"] = mqConfig.BrokerURLs
				plugin, err := mq.NewMQPlugin(mq.Kafka, config)
				if err == nil {
					// 使用类型断言
					scheduler.KafkaPlugin = plugin
					logger.Info("Kafka plugin initialized.")
				} else {
					logger.Error("Failed to initialize Kafka plugin", zap.Error(err))
				}

			case models.NSQ:
				config["nsqd_address"] = mqConfig.BrokerURLs
				plugin, err := mq.NewMQPlugin(mq.NSQ, config)
				if err == nil {
					// 使用类型断言
					scheduler.NSQPlugin = plugin
					logger.Info("NSQ plugin initialized.")
				} else {
					logger.Error("Failed to initialize NSQ plugin", zap.Error(err))
				}

			case models.RocketMQ:
				config["name_servers"] = mqConfig.BrokerURLs
				config["group_id"] = "task_queue_group"

				// 处理额外配置
				var extraMap map[string]interface{}
				if err := json.Unmarshal(mqConfig.ExtraConfig, &extraMap); err == nil {
					for k, v := range extraMap {
						config[k] = v
					}
				}

				plugin, err := mq.NewMQPlugin(mq.RocketMQ, config)
				if err == nil {
					// 使用类型断言
					scheduler.RocketMQPlugin = plugin
					logger.Info("RocketMQ plugin initialized.")
				} else {
					logger.Error("Failed to initialize RocketMQ plugin", zap.Error(err))
				}

			case models.Redis:
				config["addr"] = mqConfig.BrokerURLs

				// 处理额外配置
				var extraMap map[string]interface{}
				if err := json.Unmarshal(mqConfig.ExtraConfig, &extraMap); err == nil {
					for k, v := range extraMap {
						config[k] = v
					}
				}

				plugin, err := mq.NewMQPlugin(mq.Redis, config)
				if err == nil {
					scheduler.RedisPlugin = plugin
					logger.Info("Redis plugin initialized.")
				} else {
					logger.Error("Failed to initialize Redis plugin", zap.Error(err))
				}
			}
		}
	} else {
		logger.Error("Failed to fetch MQ configs from database", zap.Error(err))
	}

	return scheduler // 返回初始化后的调度器
}

type ReqLogin struct {
	Username string `json:"username"`
	Password string `json:"password"`
}

func Login(c *gin.Context) {
	var req ReqLogin
	if err := c.ShouldBindJSON(&req); err != nil {
		response.Fail(c, http.StatusBadRequest, "登录失败")
		return
	}
	if req.Username == "" || req.Password == "" {
		response.Fail(c, http.StatusBadRequest, "账号和密码不能为空！")
		return
	}
	username := viper.GetString("username")
	password := viper.GetString("password")
	if username != req.Username && password != req.Password {
		response.Fail(c, http.StatusBadRequest, "登录失败！请检查账号密码")
		return
	}
	token, err := utils.GenerateToken(username, password)
	if err != nil {
		response.Fail(c, http.StatusInternalServerError, "token生成失败")
		return
	}
	response.Success(c, token)

}

// initMQPlugins 初始化消息队列插件
func initMQPlugins(db *gorm.DB) map[mq.MQType]mq.MQPlugin {
	plugins := make(map[mq.MQType]mq.MQPlugin)

	// 初始化内存队列插件
	config := map[string]interface{}{
		"buffer_size":   1000,
		"default_topic": "default",
		"db":            db, // 传递数据库连接
	}

	plugin, err := mq.NewMQPlugin(mq.ChannelQueue, config)
	if err == nil {
		plugins[mq.ChannelQueue] = plugin
	}

	// 从数据库获取其他消息队列配置并初始化
	var mqConfigs []models.MQConfig
	if err := db.Find(&mqConfigs).Error; err == nil {
		for _, mqConfig := range mqConfigs {
			config := map[string]interface{}{}

			switch mq.MQType(mqConfig.PluginType) {
			case mq.RabbitMQ:
				config["conn_string"] = mqConfig.BrokerURLs
			case mq.Kafka:
				config["brokers"] = mqConfig.BrokerURLs
			case mq.NSQ:
				config["nsqd_address"] = mqConfig.BrokerURLs
			case mq.RocketMQ:
				config["name_servers"] = mqConfig.BrokerURLs
				handleExtraConfig(mqConfig.ExtraConfig, config)
			case mq.Redis:
				config["addr"] = mqConfig.BrokerURLs
				handleExtraConfig(mqConfig.ExtraConfig, config)
			default:
				continue
			}

			plugin, err := mq.NewMQPlugin(mq.MQType(mqConfig.PluginType), config)
			if err == nil {
				plugins[mq.MQType(mqConfig.PluginType)] = plugin
				log.Printf("%s 插件初始化成功", mqConfig.PluginType)
			} else {
				log.Printf("%s 插件初始化失败: %v", mqConfig.PluginType, err)
			}
		}
	}

	return plugins
}

// 处理额外配置
func handleExtraConfig(extraConfig datatypes.JSON, config map[string]interface{}) {
	var extraMap map[string]interface{}
	if err := json.Unmarshal(extraConfig, &extraMap); err == nil {
		for k, v := range extraMap {
			config[k] = v
		}
	}
}
